package com.stone.message.adapter.trigger;

import com.stone.message.adapter.domain.bo.TriggerMessageParam;
import com.stone.message.adapter.domain.event.ReceiveMessageEvent;
import com.stone.message.adapter.infra.consumer.MessageConsumer;
import com.stone.message.adapter.infra.publisher.MessagePublisher;
import com.stone.message.adapter.listener.ReceiveMessageEventListener;
import com.stone.starter.core.exception.CommonException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.Future;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * 公用抽象数据接收触发器（各种类型的触发器只用继承该类实现对应方法即可），
 * 每个获取数据来源不同的场景可以继承该类实现一个通用的公共触发器，比如：
 * 从Redis不同队列获取数据，从MongoDB获取数据，从数据库获取数据等
 *
 * @param <T> 获取到数据的类型(即要处理的数据类型)
 * @author Mr_wenpan@163.com 2021/07/14 21:55
 */
@Slf4j
public abstract class AbstractCommonReceiverTrigger<T, K> {

    /**
     * 消息发送器
     */
    private final MessagePublisher messagePublisher;

    /**
     * 注入接受者消息监听器
     * todo 这里不能直接注入ReceiveMessageEventListener，按条件检查，不是每个模块都必须要有消费者
     */
    @Autowired(required = false)
    private ReceiveMessageEventListener receiveMessageEventListener;

    /**
     * 停止标记，一般来说后台可能会启动一个守护线程，在系统关机时将该标志修改为true
     */
    private volatile boolean stopFlag;

    /**
     * 默认异步执行线程数
     */
    private final int DEFAULT_THREAD_NUM = 1;

    public AbstractCommonReceiverTrigger(MessagePublisher messagePublisher) {
        this.messagePublisher = messagePublisher;
        stopFlag = false;
    }

    /**
     * 触发执行方法
     *
     * @param messageBO  参数对象
     * @param biConsumer 消费型接口，接收两个参数，第一个是数据，第二个是TriggerMessageParam
     * @author Mr_wenpan@163.com 2021/7/15 1:45 下午
     */
    public void triggerWithCheck(TriggerMessageParam<K> messageBO, BiConsumer<T, TriggerMessageParam<K>> biConsumer) {
        // 必要的参数校验
        checkParam(messageBO);

        // 用户自定义校验
        if (!check(messageBO)) {
            log.warn("触发器校验不通过，不执行！messageBO={}", messageBO);
            return;
        }

        // 触发器执行
        triggerCommonMethod(messageBO, (param) -> asyncTriggerMessage(messageBO, biConsumer), biConsumer);

    }

    /**
     * 供应型触发方法
     *
     * @param supplier   供应型接口
     * @param biConsumer 消费型接口
     * @author Mr_wenpan@163.com 2021/7/17 2:23 下午
     */
    public void triggerWithCheck(Supplier<TriggerMessageParam<K>> supplier, BiConsumer<T, TriggerMessageParam<K>> biConsumer) {
        triggerWithCheck(supplier.get(), biConsumer);
    }

    /**
     * 同步触发消息
     *
     * @param messageBO 参数
     * @return void
     * @author Mr_wenpan@163.com 2021/7/14 10:55 下午
     */
    protected T syncTriggerMessage(TriggerMessageParam<K> messageBO, BiConsumer<T, TriggerMessageParam<K>> biConsumer) {
        // 由调用线程发送消息
        return publishMessage(messageBO, biConsumer);
    }

    /**
     * 多线程异步触发发送消息
     *
     * @param messageBO 参数
     * @return void
     * @author Mr_wenpan@163.com 2021/7/14 10:45 下午
     */
    protected List<Future<T>> asyncTriggerMessage(TriggerMessageParam<K> messageBO,
                                                  BiConsumer<T, TriggerMessageParam<K>> biConsumer) {
        // 多少个线程一起执行
        int threadNum = Objects.isNull(messageBO.getTaskCount()) ? DEFAULT_THREAD_NUM : messageBO.getTaskCount();
        List<Future<T>> futures = new ArrayList<>(threadNum);

        for (int i = 0; i < threadNum; i++) {
            futures.add(threadPoolTaskExecutor().submit(() -> publishMessage(messageBO, biConsumer)));
        }

        return futures;
    }

    /**
     * 发布消息
     *
     * @param messageParam 消息参数
     * @param biConsumer   消费型函数式接口
     * @return T
     * @author Mr_wenpan@163.com 2021/7/17 1:35 下午
     */
    private T publishMessage(TriggerMessageParam<K> messageParam, BiConsumer<T, TriggerMessageParam<K>> biConsumer) {
        boolean isFailed = false;
        T data = getData(messageParam);

        while (!stopFlag && Objects.nonNull(data)) {
            try {
                // 将消息事件发布出去
                ReceiveMessageEvent<T> receiveMessageEvent = new ReceiveMessageEvent<>(this);
                receiveMessageEvent.setContent(data);
                receiveMessageEvent.setQueue(messageParam.getQueue());
                messagePublisher.publish(receiveMessageEvent);
            } catch (Exception ex) {
                log.error("任务执行出现异常，异常线程：[{}]，异常信息：{}", Thread.currentThread().getName(), ex);
                isFailed = true;
            }
            // 如果消息发送处理失败，则进行错误补偿
            if (isFailed) {
                isFailed = false;
                // 错误补偿
                biConsumer.accept(data, messageParam);
            }
            // 不断的获取数据，直到没有数据或调用了stop
            data = getData(messageParam);
        }

        // 当stopFlag被置为true的时候，此时如果有数据取出，需要对取出的数据做补偿
        if (Objects.nonNull(data)) {
            biConsumer.accept(data, messageParam);
        }

        return data;
    }

    /**
     * 停止异步处理
     */
    public void stop() {
        stopFlag = true;
    }

    /**
     * 异步执行任务的线程池，由子类使用时动态传入
     *
     * @return org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor
     * @author Mr_wenpan@163.com 2021/7/14 10:49 下午
     */
    protected abstract ThreadPoolTaskExecutor threadPoolTaskExecutor();

    /**
     * 获取待发送的数据，由子类决定如何获取数据
     *
     * @return T
     * @author Mr_wenpan@163.com 2021/7/14 10:49 下午
     */
    protected abstract T getData(TriggerMessageParam<K> param);

    /**
     * 是否异步多线程触发执行，默认同步
     */
    protected boolean isAsyncExecute() {
        return Boolean.FALSE;
    }

    /**
     * 校验触发器必需参数
     *
     * @param messageBO 参数
     * @return boolean
     * @author Mr_wenpan@163.com 2021/7/17 1:23 下午
     */
    private boolean checkParam(TriggerMessageParam<K> messageBO) {

        // 默认实现触发器所需要的必要条件
        Assert.isTrue(Objects.isNull(messageBO), "请传递触发器触发的必要参数！");
        Assert.isTrue(StringUtils.isBlank(messageBO.getQueue()), "触发器要发送的mq队列不能为空！");

        if (Objects.nonNull(receiveMessageEventListener)) {
            // 必须保证用户所指定的queue能正确获取到一个消费者来消费数据，避免拉取数据后没有消费者消费
            MessageConsumer consumer = receiveMessageEventListener.getConsumerByQueue(messageBO.getQueue());
            if (Objects.isNull(consumer)) {
                throw new CommonException("指定的消息投递队列：[{}],找不到对应的消费者，请检查队列是否正确！");
            }
        }

        // 开启异步执行，则线程数量不能小于1，且使用的线程池不能为空
        if (isAsyncExecute()) {
            if (Objects.isNull(threadPoolTaskExecutor())) {
                throw new CommonException("多线程执行触发器需要传递具体使用的线程池");
            }
            Integer taskCount = messageBO.getTaskCount();
            if (Objects.isNull(taskCount) || taskCount <= 0) {
                throw new CommonException("多线程执行触发，线程数量必须大于0，请检查线程数量设置");
            }
        }

        return Boolean.TRUE;
    }

    /**
     * 执行前的一些必要校验，留给子类自己决定如何校验
     */
    protected boolean check(TriggerMessageParam<K> messageBO) {
        return Boolean.TRUE;
    }

    /**
     * 触发器共用方法
     *
     * @param function 函数式接口
     * @author Mr_wenpan@163.com 2021/7/15 2:14 下午
     */
    private void triggerCommonMethod(TriggerMessageParam<K> messageBO,
                                     Function<TriggerMessageParam<K>, List<Future<T>>> function,
                                     BiConsumer<T, TriggerMessageParam<K>> biConsumer) {
        // 是多线程异步执行还是同步执行由子类实现决定
        if (isAsyncExecute()) {
            List<Future<T>> futures = function.apply(messageBO);
            // 主线程等待
            for (Future<T> future : futures) {
                T result = null;
                try {
                    result = future.get();
                } catch (Exception ex) {
                    log.error("线程被抢占或中断，线程名称：[{}]，执行的数据是：{}，异常信息：{}",
                            Thread.currentThread().getName(), result, ex);
                }
            }
        } else {
            syncTriggerMessage(messageBO, biConsumer);
        }
    }
}
